package com.qwiki.xml;

import java.util.LinkedList;

import org.apache.log4j.Logger;

public abstract class PageAbstractAsyncHandler<T, S extends PageFilter<T>> extends Thread implements PageHandler<T, S> {
  protected final Logger _log = Logger.getLogger(this.getClass());
  LinkedList<T> queue = new LinkedList<T>();
  private final String name;
  private final int id;
  private final int queueSize;
  private final Object lock;
  private volatile boolean running;
  private volatile boolean isShutdown;
  private final int block = 100;
  private long last = System.currentTimeMillis();
  private int pagecount;


  public PageAbstractAsyncHandler(String name, int id, int queueSize) {
    super(name + " - " + id);
    this.name = name;
    this.id = id;
    lock = new String("lock " + name + " - " + id);
    this.queueSize = queueSize;
  }

  public PageAbstractAsyncHandler(String name, int id) {
    this(name, id, 2000);
  }

  @Override
  public void start() {
    running = true;
    isShutdown = false;
    super.start();
  }

  public void drain() {
    synchronized (lock) {
      while (!queue.isEmpty()) {
        try {
          lock.wait(200l);
        } catch (InterruptedException e) {
          Thread.interrupted();
          _log.warn(e);
        }
      }
      running = false;
    }
  }

  public void shutdown() {
    running = false;
  }

  public boolean isRunning() {
    return !isShutdown;
  }

  @Override
  public void run() {
    _log.info(getName() + " start processing");
    outer:
    while (running) {
      T page = null;
      synchronized (lock) {
        while (queue.isEmpty()) {
          if (!running)
            break outer;
          try {
            lock.wait(200l);
          } catch (InterruptedException e) {
            Thread.interrupted();
            _log.warn(e);
          }
        }
        page = queue.poll();
      }
      pagecount++;
      if (pagecount % block == 0) {
        long now = System.currentTimeMillis();
        long elapsed = now - last;
        if (elapsed <= 1)
          elapsed = 1;
        last = now;
        int speedPerS = (int) (1000l * block / elapsed);
        int speedPerH = (int) (60l * 60l * 1000l * block / elapsed);
        String msg = pagecount + " pages processed at " + speedPerS + " p/s " + speedPerH + " p/h per thread";
        System.out.println(msg);
        _log.info(msg);
      }
      doProcess(page);
      page = null;
    }
    isShutdown = true;
    _log.info(getName() + " end processing and exit");
  }

  public abstract void doProcess(T page);

  @Override
  public boolean process(T page) {
    synchronized (lock) {
      // wait until queue is not full
      while (queue.size() > queueSize) {
        try {
          lock.wait(200l);
        } catch (InterruptedException e) {
          Thread.interrupted();
          _log.warn(e);
        }
      }
      queue.add(page);
      lock.notifyAll();
    }
    return true;
  }

}
